package org.gruposp2p.cometd;

import java.util.Iterator;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.cometd.bayeux.Message;
import org.cometd.bayeux.server.ServerChannel;
import org.cometd.bayeux.server.ServerSession;
import org.gruposp2p.proyectokarma.domain.CometdMessage;
import com.infusion.util.event.EventBroker;
import com.infusion.util.event.EventConsumer;
import org.codehaus.groovy.grails.commons.ApplicationHolder;
import org.cometd.bayeux.client.ClientSessionChannel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.context.ApplicationContext;

public class AlertChannelMessageProcessor implements EventConsumer {

    private static Logger logger =
            LoggerFactory.getLogger(AlertChannelMessageProcessor.class);

    private final ConcurrentMap<String, String> members =
		new ConcurrentHashMap<String, String>();
    public static final String CHANNEL_NAME = "/alerts";
    KarmaMessageService karmaMessageService;
    CometdMessageService messageService;
    ServerChannel channel;

    public AlertChannelMessageProcessor (KarmaMessageService karmaMessageService) {
        this.karmaMessageService = karmaMessageService;
        ApplicationContext context = (ApplicationContext)ApplicationHolder.
                getApplication().getMainContext();
        EventBroker eventBroker = (EventBroker) context.getBean("eventBroker");
        eventBroker.subscribe(KarmaMessageEvent.ALERT_CHANNEL, this);
        messageService = (CometdMessageService) context.getBean("cometdMessageService");
        channel = karmaMessageService.getBayeux().getChannel(CHANNEL_NAME);
        logger.debug("Subscribed to Event: " + KarmaMessageEvent.ALERT_CHANNEL);
    }

    public void processClientMessage (ServerSession remoteClient, Message message) {
        logger.debug(" processClientMessage - message: " + message +
                " - remoteClient: " + remoteClient + " - user agent: " +
                remoteClient.getUserAgent());
        CometdMessage cometdMessage =
                messageService.getCometdMessageFromJSON(message.getJSON());
        if (CometdMessage.ACTION_JOIN.equals(cometdMessage.getAction())) {
            processAlertJoin(remoteClient, cometdMessage);
        }
    	remoteClient.deliver(karmaMessageService.getServerSession(),
                message.getChannel(), message, null);
        ServerChannel serverChannel = karmaMessageService.getBayeux().
                getChannel(message.getChannel());
        Set<ServerSession> subscribers = (Set<ServerSession>) serverChannel.getSubscribers();
        for (ServerSession subscriber : subscribers) {
            logger.debug(" - Subscriber: " + subscriber);
        }
        logger.debug(" - getId(): " + serverChannel.getId() +
                " - serverChannel.isPersistent(): " + serverChannel.isPersistent() +
                " - isBroadcast: " + serverChannel.isBroadcast() +
                " - isService(): " + serverChannel.isService());
        serverChannel.publish(karmaMessageService.getServerSession(),
                "gromenaguer0", message.getId());
        remoteClient.deliver(karmaMessageService.getServerSession(), "/service/alert",
                "gromenaguer1", "server" + message.getId());
    }

    private void processAlertJoin (final ServerSession remoteClient,
            final CometdMessage message) {
        logger.debug(" - processJoin - ");
        if (members.containsKey(message.getPeerFrom().getUsername())) return;
        members.put(message.getPeerFrom().getUsername(), remoteClient.getId());
        if (!karmaMessageService.getOort().isOort(remoteClient))
                    karmaMessageService.getSeti().associate(
                            message.getPeerFrom().getUsername(),remoteClient);
        remoteClient.addListener(new ServerSession.RemoveListener() {
            public void removed(ServerSession session, boolean timeout) {
                logger.debug(" - remoteClient: " + session.getId()
                        + " removed from alerts");
                if (!karmaMessageService.getOort().isOort(remoteClient))
                            karmaMessageService.getSeti().disassociate(
                                    message.getPeerFrom().getUsername());
                members.values().remove(session.getId());
            }
        });
    }

    public void consume(Object event, EventBroker eventBroker) {
        KarmaMessageEvent messageEvent = ((KarmaMessageEvent)event);
        String cometdMessage = messageService.getJSONFromCometdMessage(
                messageEvent.getMessage());
        logger.debug(" - consume - CometdMessage: " + cometdMessage);
        ClientSessionChannel channel =
                karmaMessageService.getLocalSession().getChannel(CHANNEL_NAME);
        channel.publish(cometdMessage);
    }

}
